package com.bw.gmall.realtime.Day0911;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class KafkaProducer {
    public static void main(String[] args) throws Exception {
        // todo 创建流式执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度为1
        env.setParallelism(1);

        // todo Kafka的topic主题名称
        String topic_name = "topic_log" ;

        DataStreamSource<String> log_data = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic_name, "base_consumer_log").setStartFromEarliest());

//        log_data.print();

        System.out.println("==============================================================");
        SingleOutputStreamOperator<JSONObject> start_value = log_data.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {

                // todo 判断日志数据是否为空
                if (value != null && !value.equals("")) {
                    // 将String格式的日志数据转换为JSON格式
                    JSONObject json = JSONObject.parseObject(value);
                    // 判断转换后的JSON格式是否正确
                    if (json != null) {
                        // 创建一个新的JSON对象，用来接收启动日志数据
                        JSONObject new_json = new JSONObject();
                        new_json.put("common", json.getJSONObject("common"));
                        // 将时间戳以长整型类型取值并放入新的JSON中
                        new_json.put("ts", json.getLong("ts"));
                        // 判断启动日志数据是否为空
                        if (json.getJSONObject("start") != null) {
                            new_json.put("start", json.getJSONObject("start"));
                            out.collect(new_json);
                        }
                    }
                }

            }
        });

        start_value.print("启动日志数据----->");

        System.out.println("==============================================================");
        SingleOutputStreamOperator<JSONObject> page_value = log_data.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                JSONObject json = JSONObject.parseObject(value);

                JSONObject new_json = new JSONObject();

                if (json != null) {
                    new_json.put("common", json.getJSONObject("common"));
                    new_json.put("ts", json.getLong("ts"));
                    if (json.getJSONObject("page") != null) {
                        new_json.put("page", json.getJSONObject("page"));
                        out.collect(new_json);
                    }
                }
            }
        });

        page_value.print("页面日志数据----->");

        System.out.println("==============================================================");
        SingleOutputStreamOperator<JSONObject> display_data = log_data.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                try {
                    if (value != null){
                        JSONObject json = JSONObject.parseObject(value);

                        JSONObject new_json = new JSONObject();

                         if (json != null) {
                            new_json.put("common",json.getJSONObject("common"));
                            new_json.put("ts",json.getLong("ts"));

                            JSONArray displays = json.getJSONArray("displays");
                            if (displays!=null && displays.size() > 0 ) {
                                for (int i = 0; i < displays.size(); i++) {
                                    new_json.put("display",displays.getJSONObject(i));
                                    out.collect(new_json);
                                }
                            }
                        }
                    }
                } catch (Exception e) {

                }
            }
        });

        display_data.print("曝光日志数据----->");

        System.out.println("==============================================================");

        SingleOutputStreamOperator<JSONObject> action_data = log_data.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                try {
                    if (value != null){
                        JSONObject json = JSONObject.parseObject(value);

                        JSONObject new_json = new JSONObject();

                        if (json != null) {
                            new_json.put("common",json.getJSONObject("common"));
                            new_json.put("ts",json.getLong("ts"));

                            JSONArray actions = json.getJSONArray("action");
                            if (actions!=null && actions.size() > 0 ) {
                                for (int i = 0; i < actions.size(); i++) {
                                    new_json.put("action",actions.getJSONObject(i));
                                    out.collect(new_json);
                                }
                            }
                        }
                    }
                } catch (Exception e) {

                }
            }
        });

        action_data.print("动作日志数据----->");

        // todo 执行
        env.execute();
    }
}
